package com.kafka.kafkademo;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

/**
 * @author liyanhao
 * @Date 2023-02-18 17:01
 */
public class MsgConsumer {
    private final static String TOPIC_NAME = "my-replicated-topic";
    private final static String CONSUMER_GROUP_NAME = "testGroup";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.139.30:9092");
        //消费分组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
        //是否自动提价offset 默认就是true
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        //自动提交offset的间隔时间
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        //consumer给broker发送心跳的时间间隔
        properties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
        //服务端broker多久感知不到一个consumer心跳就认为他故障了，会将其提出消费组
        //对应partition也会重新分配给其他consumer 默认是10秒
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);
        //一次poll最大拉取消息的条数，如果消费者处理速度很快，可以设置大点，如果处理速度一般，可以设置小点
        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        //如果两次poll操作间隔超过这个时间，broker就会任务这个consumer处理能力太弱，会将其踢出消费组，将分区分配给别的消费consumer消费
        properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        //自动订阅分区消费
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        //消费指定分区
       // consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        //消息回溯消费
//        consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
//        consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));

        //指定offset消费
//        consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
//        consumer.seek(new TopicPartition(TOPIC_NAME, 0), 0);
        //从指定时间开始消费
        List<PartitionInfo> partitionInfos = consumer.partitionsFor(TOPIC_NAME);
        //从一小时前开始消费
        long fetchDateTime = new Date().getTime() - 1000 * 60 * 60;
        HashMap<TopicPartition, Long> map = new HashMap<>();
        for (PartitionInfo par : partitionInfos) {
            map.put(new TopicPartition(TOPIC_NAME, par.partition()), fetchDateTime);
        }

        Map<TopicPartition, OffsetAndTimestamp> parmap = consumer.offsetsForTimes(new HashMap<>());
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parmap.entrySet()) {
            TopicPartition key = entry.getKey();
            OffsetAndTimestamp value = entry.getValue();
            if (key == null || value == null) {
                continue;
            }
            System.out.println("partition-" + key.partition() + "|offset-" + value.offset());
            System.out.println();
            //根据消费里的timestamp确定offset
            if (value != null) {
                consumer.assign(Arrays.asList(key));
                consumer.seek(key, value.offset());
            }
        }
        while (true) {
            //poll API时拉去消息的长轮询
            ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("收到消息：partition = %d,offset = %d, key = %s, value = %s%n", record.partition(),
                        record.offset(), record.key(), record.value());
            }
            if (records.count() > 0) {
                //手动同步提交offset 当前线程会阻塞知道offset提交成功
                //一般使用同步提交 因为提交之后一般也没有什么逻辑代码了
                consumer.commitAsync();
                //手动异步提交offset 当前线程提交offset不会阻塞 可继续处理后面的程序逻辑
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
                        if (e != null) {
                            System.err.println("Commit failed for " + offsets);
                            System.err.println("Commit failed exception: " + e.getStackTrace());
                        }
                    }
                });
            }
        }
    }


}



















